#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Redis 连接管理模块

提供：
- Redis 连接池管理
- 统一的缓存操作接口
- 异常处理与日志记录
- 重试机制和降级策略
- 缓存雪崩防护
"""

import json
import random
import threading
from contextlib import contextmanager
from datetime import datetime
from typing import Optional, Any, Dict, List

import redis
from redis.connection import ConnectionPool
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

from common.config import settings
from common.logger import create_logger

logger = create_logger(__name__)


class RedisManager:
	"""Redis 连接管理器 - 单例模式，带降级策略"""

	_instance = None
	_lock = threading.RLock()
	_pool: Optional[ConnectionPool] = None
	_client: Optional[redis.Redis] = None
	_available = False  # Redis可用性标志

	def __new__(cls):
		if cls._instance is None:
			with cls._lock:
				if cls._instance is None:
					cls._instance = super().__new__(cls)
		return cls._instance

	def __init__(self):
		if not hasattr(self, '_initialized'):
			self._initialized = True
			self._init_connection()

	@retry(
		stop=stop_after_attempt(3),
		wait=wait_exponential(multiplier=1, min=2, max=8),
		retry=retry_if_exception_type((redis.ConnectionError, redis.TimeoutError))
	)
	def _init_connection(self):
		"""初始化Redis连接池，带重试机制"""
		try:
			# 从配置中读取Redis参数
			logger.info(f"初始化Redis连接: {settings.redis_host}:{settings.redis_port}")

			# 创建连接池
			self._pool = ConnectionPool(
				host=settings.redis_host,
				port=settings.redis_port,
				password=settings.redis_password,
				db=settings.redis_db,
				decode_responses=True,
				max_connections=settings.redis_max_connections,
				socket_timeout=settings.redis_socket_timeout,
				socket_connect_timeout=settings.redis_socket_connect_timeout,
				retry_on_timeout=True,
				health_check_interval=settings.redis_health_check_interval
			)
			self._client = redis.Redis(connection_pool=self._pool)

			# 测试连接
			self._client.ping()
			self._available = True
			logger.info(f"Redis连接池初始化成功 [环境: {settings.env}]")

		except Exception as e:
			logger.error(f"Redis连接初始化失败: {e}")
			self._client = None
			self._pool = None
			self._available = False
			# 不再抛出异常，允许降级到数据库
			logger.warning("Redis不可用，系统将降级使用数据库")

	def get_client(self) -> Optional[redis.Redis]:
		"""获取Redis客户端，可能返回None"""
		if not self._available:
			return None

		if self._client is None:
			self._init_connection()

		return self._client if self._available else None

	def is_connected(self) -> bool:
		"""检查Redis连接状态"""
		if not self._available:
			return False

		try:
			if self._client:
				self._client.ping()
				return True
		except Exception as e:
			logger.warning(f"Redis连接检查失败: {e}")
			self._available = False
		return False

	def reconnect(self):
		"""重新连接Redis"""
		logger.info("尝试重新连接Redis...")
		self._init_connection()


class RedisService:
	"""Redis服务封装类，支持降级策略"""

	def __init__(self):
		self.manager = RedisManager()

	@property
	def client(self) -> Optional[redis.Redis]:
		"""获取Redis客户端"""
		return self.manager.get_client()

	def is_available(self) -> bool:
		"""检查Redis是否可用"""
		return self.manager.is_connected()

	def set(self, key: str, value: Any, ex: Optional[int] = None, nx: bool = False) -> bool:
		"""
		设置键值对，带缓存雪崩防护和去重支持

		Args:
			key: 键名
			value: 值（自动序列化为JSON）
			ex: 过期时间（秒）
			nx: 是否仅在key不存在时设置（SET IF NOT EXISTS）

		Returns:
			bool: 是否设置成功
		"""
		try:
			client = self.client
			if not client:
				logger.debug("Redis不可用，跳过缓存设置")
				return False

			value = self._serialize_value(value)

			if isinstance(value, (dict, list)):
				value = json.dumps(value, ensure_ascii=False)

			# 添加随机抖动防止缓存雪崩（仅在非NX模式下）
			if ex is not None and ex > 0 and not nx:
				jitter = random.randint(0, min(60, max(1, ex // 10)))  # 最多10%的抖动，最少1秒
				ex = ex + jitter

			# 使用原子性的SET命令，支持NX和EX参数
			result = client.set(key, value, ex=ex, nx=nx)
			return bool(result)
		except Exception as e:
			logger.error(f"Redis SET操作失败 key={key}, nx={nx}, ex={ex}: {e}")
			self.manager._available = False  # 标记Redis不可用
			return False

	def get(self, key: str) -> Optional[Any]:
		"""
		获取键值，支持降级

		Args:
			key: 键名

		Returns:
			Any: 值（自动反序列化JSON），Redis不可用时返回None
		"""
		try:
			client = self.client
			if not client:
				logger.debug(f"Redis不可用，无法获取缓存 key={key}")
				return None

			value = client.get(key)
			if value is None:
				return None

			# 尝试解析JSON
			try:
				if isinstance(value, str):
					return json.loads(value)
				return value
			except (json.JSONDecodeError, TypeError):
				return value

		except Exception as e:
			logger.error(f"Redis GET操作失败 key={key}: {e}")
			self.manager._available = False  # 标记Redis不可用
			return None

	def delete(self, *keys: str) -> int:
		"""
		删除键

		Args:
			keys: 键名列表

		Returns:
			int: 删除的键数量
		"""
		try:
			client = self.client
			if not client:
				return 0

			result = client.delete(*keys)
			return int(result) if result is not None else 0
		except Exception as e:
			logger.error(f"Redis DELETE操作失败 keys={keys}: {e}")
			self.manager._available = False
			return 0

	def exists(self, key: str) -> bool:
		"""
		检查键是否存在

		Args:
			key: 键名

		Returns:
			bool: 是否存在
		"""
		try:
			client = self.client
			if not client:
				return False

			result = client.exists(key)
			return bool(result)
		except Exception as e:
			logger.error(f"Redis EXISTS操作失败 key={key}: {e}")
			self.manager._available = False
			return False

	def expire(self, key: str, seconds: int) -> bool:
		"""
		设置键过期时间

		Args:
			key: 键名
			seconds: 过期时间（秒）

		Returns:
			bool: 是否设置成功
		"""
		try:
			client = self.client
			if not client:
				return False

			result = client.expire(key, seconds)
			return bool(result)
		except Exception as e:
			logger.error(f"Redis EXPIRE操作失败 key={key}: {e}")
			self.manager._available = False
			return False

	def ttl(self, key: str) -> int:
		"""
		获取键剩余过期时间

		Args:
			key: 键名

		Returns:
			int: 剩余秒数，-1表示永不过期，-2表示不存在
		"""
		try:
			client = self.client
			if not client:
				return -2

			result = client.ttl(key)
			return int(result) if result is not None else -2
		except Exception as e:
			logger.error(f"Redis TTL操作失败 key={key}: {e}")
			self.manager._available = False
			return -2

	def incr(self, key: str, amount: int = 1) -> Optional[int]:
		"""
		递增计数器

		Args:
			key: 键名
			amount: 递增量

		Returns:
			int: 递增后的值
		"""
		try:
			client = self.client
			if not client:
				return None

			result = client.incr(key, amount)
			return int(result) if result is not None else None
		except Exception as e:
			logger.error(f"Redis INCR操作失败 key={key}: {e}")
			self.manager._available = False
			return None

	def hset(self, name: str, mapping: Dict[str, Any]) -> int:
		"""
		设置哈希表字段

		Args:
			name: 哈希表名
			mapping: 字段映射

		Returns:
			int: 新增字段数量
		"""
		try:
			client = self.client
			if not client:
				return 0

			# 序列化复杂对象
			serialized_mapping = {}
			for k, v in mapping.items():
				if isinstance(v, (dict, list)):
					serialized_mapping[k] = json.dumps(v, ensure_ascii=False)
				else:
					serialized_mapping[k] = v
			result = client.hset(name, mapping=serialized_mapping)
			return int(result) if result is not None else 0
		except Exception as e:
			logger.error(f"Redis HSET操作失败 name={name}: {e}")
			self.manager._available = False
			return 0

	def hget(self, name: str, key: str) -> Optional[Any]:
		"""
		获取哈希表字段值

		Args:
			name: 哈希表名
			key: 字段名

		Returns:
			Any: 字段值
		"""
		try:
			client = self.client
			if not client:
				return None

			value = client.hget(name, key)
			if value is None:
				return None

			# 尝试解析JSON
			try:
				if isinstance(value, str):
					return json.loads(value)
				return value
			except (json.JSONDecodeError, TypeError):
				return value

		except Exception as e:
			logger.error(f"Redis HGET操作失败 name={name} key={key}: {e}")
			self.manager._available = False
			return None

	def hgetall(self, name: str) -> Dict[str, Any]:
		"""
		获取哈希表所有字段

		Args:
			name: 哈希表名

		Returns:
			Dict: 所有字段映射
		"""
		try:
			client = self.client
			if not client:
				return {}

			result = client.hgetall(name)
			if not isinstance(result, dict):
				return {}

			# 尝试反序列化JSON值
			processed_result = {}
			for k, v in result.items():
				try:
					if isinstance(v, str):
						processed_result[k] = json.loads(v)
					else:
						processed_result[k] = v
				except (json.JSONDecodeError, TypeError):
					processed_result[k] = v
			return processed_result
		except Exception as e:
			logger.error(f"Redis HGETALL操作失败 name={name}: {e}")
			self.manager._available = False
			return {}

	def _serialize_value(self, value: Any) -> str:
		"""递归地将 datetime 转为 ISO 格式字符串"""
		if isinstance(value, datetime):
			return value.isoformat()
		elif isinstance(value, dict):
			return {k: self._serialize_value(v) for k, v in value.items()}
		elif isinstance(value, list):
			return [self._serialize_value(v) for v in value]
		else:
			return value

	# ============ 序列号管理功能 ============

	def eval_lua_script(self, script: str, keys: List[str], args: List[Any]) -> Any:
		"""
		执行Lua脚本

		Args:
			script: Lua脚本内容
			keys: Redis键列表
			args: 参数列表

		Returns:
			脚本执行结果
		"""
		try:
			client = self.client
			if not client:
				logger.debug("Redis不可用，无法执行Lua脚本")
				return None

			result = client.eval(script, len(keys), *keys, *args)
			return result

		except Exception as e:
			logger.error(f"执行Lua脚本失败: {e}")
			self.manager._available = False
			return None

	def atomic_incr_with_limit(self, key: str, increment: int, max_value: int, warning_threshold: float = 0.9) -> \
			Optional[Dict[str, Any]]:
		"""
		原子递增操作，带上限检查和预警

		Args:
			key: Redis键名
			increment: 递增量
			max_value: 最大值限制
			warning_threshold: 预警阈值（默认90%）

		Returns:
			{
				"success": bool,
				"start_value": int,  # 起始值
				"end_value": int,    # 结束值
				"warning": bool      # 是否触发预警
			}
		"""
		lua_script = """
		local key = KEYS[1]
		local increment = tonumber(ARGV[1])
		local max_val = tonumber(ARGV[2])
		local warning_threshold = tonumber(ARGV[3])

		local current = redis.call('GET', key)
		if not current then
			current = 0
		else
			current = tonumber(current)
		end

		-- 检查是否超过上限
		if current + increment > max_val then
			return {0, current, current, 0}  -- success=0, start, end, warning
		end

		-- 执行递增
		local new_val = redis.call('INCRBY', key, increment)

		-- 检查预警
		local warning_val = math.floor(max_val * warning_threshold)
		local warning_flag = (new_val >= warning_val) and 1 or 0

		-- 返回结果：success, start_value, end_value, warning
		local start_val = new_val - increment + 1
		return {1, start_val, new_val, warning_flag}
		"""

		try:
			result = self.eval_lua_script(lua_script, [key], [increment, max_value, warning_threshold])

			if not result:
				return None

			success, start_value, end_value, warning = result

			return {
				"success": bool(success),
				"start_value": int(start_value),
				"end_value": int(end_value),
				"warning": bool(warning)
			}

		except Exception as e:
			logger.error(f"原子递增操作失败 key={key}: {e}")
			return None

	def get_sequence_batch(self, pool_key: str, batch_size: int, max_value: int) -> Optional[List[int]]:
		"""
		批量获取序列号

		Args:
			pool_key: 序列号池键名
			batch_size: 批量大小
			max_value: 最大值限制

		Returns:
			序列号列表，失败返回None
		"""
		result = self.atomic_incr_with_limit(pool_key, batch_size, max_value)

		if not result or not result["success"]:
			return None

		start_val = result["start_value"]
		end_val = result["end_value"]

		# 生成序列号列表
		sequences = list(range(start_val, end_val + 1))

		# 记录预警
		if result["warning"]:
			logger.warning(f"序列号池使用量已达预警阈值: {pool_key}, 当前值: {end_val}/{max_value}")

		return sequences

	def reset_daily_pools(self, area_codes: List[str], date_str: Optional[str] = None) -> Dict[str, bool]:
		"""
		重置日级流水号池（台区编码用）

		Args:
			area_codes: 区县码列表
			date_str: 日期字符串（YYYYMMDD），默认为当前日期

		Returns:
			重置结果字典 {pool_key: success}
		"""
		if not date_str:
			from datetime import date
			date_str = date.today().strftime("%Y%m%d")

		results = {}

		for area_code in area_codes:
			pool_key = f"station:{area_code}:{date_str}"

			try:
				# 重置为0
				success = self.set(pool_key, 0, ex=24 * 60 * 60)  # 24小时过期
				results[pool_key] = success

				if success:
					logger.info(f"重置日级流水号池成功: {pool_key}")
				else:
					logger.warning(f"重置日级流水号池失败: {pool_key}")

			except Exception as e:
				logger.error(f"重置日级流水号池异常: {pool_key}, error: {e}")
				results[pool_key] = False

		return results

	def get_pool_status(self, pool_keys: List[str]) -> Dict[str, Dict[str, Any]]:
		"""
		获取多个序列号池的状态

		Args:
			pool_keys: 池键名列表

		Returns:
			{
				pool_key: {
					"current_value": int,
					"exists": bool,
					"ttl": int
				}
			}
		"""
		results = {}

		for pool_key in pool_keys:
			try:
				current_value = self.get(pool_key)
				exists = self.exists(pool_key)
				ttl = self.ttl(pool_key)

				results[pool_key] = {
					"current_value": int(current_value) if current_value else 0,
					"exists": exists,
					"ttl": ttl
				}

			except Exception as e:
				logger.error(f"获取池状态失败: {pool_key}, error: {e}")
				results[pool_key] = {
					"current_value": 0,
					"exists": False,
					"ttl": -2
				}

		return results


# 全局Redis服务实例
redis_service = RedisService()


def get_redis_service() -> RedisService:
	"""获取Redis服务实例"""
	return redis_service


@contextmanager
def redis_fallback(operation_name: str = "Redis操作"):
	"""
	Redis操作的上下文管理器，提供降级处理

	Usage:
		with redis_fallback("获取KPI数据"):
			data = redis_service.get("kpi_key")
			if data is None:
				# 降级到数据库查询
				data = query_from_database()
	"""
	try:
		yield
	except Exception as e:
		logger.warning(f"{operation_name}失败，已降级: {e}")
		# 这里可以添加降级逻辑，比如从数据库查询
		pass
